/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.consumer;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.common.message.MessageConst;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.body.ProcessQueueInfo;

/**
 * Queue consumption snapshot
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-24
 */
public class ProcessQueue {
	public final static long RebalanceLockMaxLiveTime = Long.parseLong(System.getProperty("rocketmq.client.rebalance.lockMaxLiveTime", "30000"));
	public final static long RebalanceLockInterval = Long.parseLong(System.getProperty("rocketmq.client.rebalance.lockInterval", "20000"));

	private final Logger log = ClientLogger.getLog();
	private final ReadWriteLock lockTreeMap = new ReentrantReadWriteLock();
	private final TreeMap<Long, MessageExt> msgTreeMap = new TreeMap<Long, MessageExt>();
	private volatile long queueOffsetMax = 0L;
	private final AtomicLong msgCount = new AtomicLong();

	private volatile boolean dropped = false;
	private volatile long lastPullTimestamp = System.currentTimeMillis();
	private final static long PullMaxIdleTime = Long.parseLong(System.getProperty("rocketmq.client.pull.pullMaxIdleTime", "120000"));

	private volatile long lastConsumeTimestamp = System.currentTimeMillis();

	private final Lock lockConsume = new ReentrantLock();

	private volatile boolean locked = false;
	private volatile long lastLockTimestamp = System.currentTimeMillis();
	private volatile boolean consuming = false;
	private final TreeMap<Long, MessageExt> msgTreeMapTemp = new TreeMap<Long, MessageExt>();
	private final AtomicLong tryUnlockTimes = new AtomicLong(0);

	private volatile long msgAccCnt = 0;

	public boolean isLockExpired() {
		boolean result = (System.currentTimeMillis() - this.lastLockTimestamp) > RebalanceLockMaxLiveTime;
		return result;
	}

	public boolean isPullExpired() {
		boolean result = (System.currentTimeMillis() - this.lastPullTimestamp) > PullMaxIdleTime;
		return result;
	}

	public boolean putMessage(final List<MessageExt> msgs) {
		boolean dispatchToConsume = false;
		try {
			this.lockTreeMap.writeLock().lockInterruptibly();
			try {
				int validMsgCnt = 0;
				for (MessageExt msg : msgs) {
					// A）遍历List<MessageExt>列表，以每个MessageExt对象的queueOffset值为key值，将MessageExt对象存入msgTreeMap:TreeMap<Long, MessageExt>变量中；该变量类型根据key值大小排序
					MessageExt old = msgTreeMap.put(msg.getQueueOffset(), msg);
					if (null == old) {
						validMsgCnt++;
						this.queueOffsetMax = msg.getQueueOffset();
					}
				}

				// B）更新ProcessQueue.msgCount变量，记录消息个数；
				msgCount.addAndGet(validMsgCnt);

				// C）经过第A步处理之后，
				// 若msgTreeMap变量不是空并且ProcessQueue.consuming为false（初始化为false）
				// 则置consuming为true（在该msgTreeMap变量消费完之后再置为false）、置临时变量dispatchToConsume为true；
				// 否则置临时变量dispatchToConsume为false
				// 表示没有待消费的消息或者msgTreeMap变量中存入了数据还未消费完，在没有消费完之前不允许在此提交消费请求，在消费完msgTreeMap之后置consuming为false；
				if (!msgTreeMap.isEmpty() && !this.consuming) {
					dispatchToConsume = true;
					this.consuming = true;
				}

				if (!msgs.isEmpty()) {
					// D）取List<MessageExt>列表的最后一个MessageExt对象
					MessageExt messageExt = msgs.get(msgs.size() - 1);
					String property = messageExt.getProperty(MessageConst.PROPERTY_MAX_OFFSET);
					if (property != null) {
						// 该对象的properties属性中取MAX_OFFSET的值，减去该MessageExt对象的queueOffset值，即为Broker端该topic和queueId下消息队列中未消费的逻辑队列的大小，
						long accTotal = Long.parseLong(property) - messageExt.getQueueOffset();
						if (accTotal > 0) {
							// 将该值存入ProcessQueue.msgAccCnt变量，用于MQClientInstance. adjustThreadPool()方法动态调整线程池大小（在MQClientInstance中启动定时任务定时调整线程池大小）
							this.msgAccCnt = accTotal;
						}
					}
				}
			} finally {
				this.lockTreeMap.writeLock().unlock();
			}
		} catch (InterruptedException e) {
			log.error("putMessage exception", e);
		}

		// E）返回临时变量dispatchToConsume值；
		return dispatchToConsume;
	}

	public long getMaxSpan() {
		try {
			this.lockTreeMap.readLock().lockInterruptibly();
			try {
				if (!this.msgTreeMap.isEmpty()) {
					return this.msgTreeMap.lastKey() - this.msgTreeMap.firstKey();
				}
			} finally {
				this.lockTreeMap.readLock().unlock();
			}
		} catch (InterruptedException e) {
			log.error("getMaxSpan exception", e);
		}

		return 0;
	}

	public long removeMessage(final List<MessageExt> msgs) {
		long result = -1;
		final long now = System.currentTimeMillis();
		try {
			this.lockTreeMap.writeLock().lockInterruptibly();
			this.lastConsumeTimestamp = now;
			try {
				if (!msgTreeMap.isEmpty()) {
					result = this.queueOffsetMax + 1;
					int removedCnt = 0;
					for (MessageExt msg : msgs) {
						MessageExt prev = msgTreeMap.remove(msg.getQueueOffset());
						if (prev != null) {
							removedCnt--;
						}
					}
					msgCount.addAndGet(removedCnt);

					if (!msgTreeMap.isEmpty()) {
						result = msgTreeMap.firstKey();
					}
				}
			} finally {
				this.lockTreeMap.writeLock().unlock();
			}
		} catch (Throwable t) {
			log.error("removeMessage exception", t);
		}

		return result;
	}

	public TreeMap<Long, MessageExt> getMsgTreeMap() {
		return msgTreeMap;
	}

	public AtomicLong getMsgCount() {
		return msgCount;
	}

	public boolean isDropped() {
		return dropped;
	}

	public void setDropped(boolean dropped) {
		this.dropped = dropped;
	}

	public void setLocked(boolean locked) {
		this.locked = locked;
	}

	public boolean isLocked() {
		return locked;
	}

	public void rollback() {
		try {
			this.lockTreeMap.writeLock().lockInterruptibly();
			try {
				this.msgTreeMap.putAll(this.msgTreeMapTemp);
				this.msgTreeMapTemp.clear();
			} finally {
				this.lockTreeMap.writeLock().unlock();
			}
		} catch (InterruptedException e) {
			log.error("rollback exception", e);
		}
	}

	public long commit() {
		try {
			// A）获取ProcessQueue.lockTreeMap:ReentrantReadWriteLock锁对该方法的整个处理逻辑加锁；
			this.lockTreeMap.writeLock().lockInterruptibly();
			try {
				// B）从msgTreeMapTemp（在从msgTreeMap中获取消息并消费时存入的）中获取最后一个元素的key值，即最大的offset；
				Long offset = this.msgTreeMapTemp.lastKey();

				// C）将ProcessQueue.msgCount值减去临时变量msgTreeMapTemp中的个数，即表示剩下的未消费的消息个数；
				msgCount.addAndGet(this.msgTreeMapTemp.size() * (-1));

				// D）清空临时变量msgTreeMapTemp列表的值；
				this.msgTreeMapTemp.clear();
				if (offset != null) {
					// E)返回最大的offset+1的值并赋值给局部变量commitOffset值用于更新消费进度之用；
					return offset + 1;
				}
			} finally {
				this.lockTreeMap.writeLock().unlock();
			}
		} catch (InterruptedException e) {
			log.error("commit exception", e);
		}

		return -1;
	}

	public void makeMessageToCosumeAgain(List<MessageExt> msgs) {
		try {
			// A）获取ProcessQueue.lockTreeMap:ReentrantReadWriteLock锁对该方法的整个处理逻辑加锁；
			this.lockTreeMap.writeLock().lockInterruptibly();
			try {
				// B）将List<MessageExt>列表的对象重新放入msgTreeMap变量中
				// 遍历List<MessageExt>列表的每个MessageExt对象
				for (MessageExt msg : msgs) {
					// 以该对象的queueoffset值为key值从msgTreeMapTemp中删除对应的MessageExt对象
					this.msgTreeMapTemp.remove(msg.getQueueOffset());
					// 将MessageExt对象以MessageExt对象的queueoffset为key值重新加入到ProcessQueue.msgTreeMap变量中；
					this.msgTreeMap.put(msg.getQueueOffset(), msg);
				}
			} finally {
				this.lockTreeMap.writeLock().unlock();
			}
		} catch (InterruptedException e) {
			log.error("makeMessageToCosumeAgain exception", e);
		}
	}

	public List<MessageExt> takeMessags(final int batchSize) {
		List<MessageExt> result = new ArrayList<MessageExt>(batchSize);
		final long now = System.currentTimeMillis();
		try {
			this.lockTreeMap.writeLock().lockInterruptibly();
			this.lastConsumeTimestamp = now;
			try {
				if (!this.msgTreeMap.isEmpty()) {
					// 从ProcessQueue.msgTreeMap变量中获取batchSize个数的List<MessageExt>列表；并且从msgTreeMap中删除
					for (int i = 0; i < batchSize; i++) {
						Map.Entry<Long, MessageExt> entry = this.msgTreeMap.pollFirstEntry();
						if (entry != null) {
							result.add(entry.getValue());
							// 存入临时变量msgTreeMapTemp中
							msgTreeMapTemp.put(entry.getKey(), entry.getValue());
						} else {
							break;
						}
					}
				}

				if (result.isEmpty()) {
					// 若该列表为空，表示该msgTreeMap列表中的消息已经消费完了，置ProcessQueue.consuming等于false；
					consuming = false;
				}
			} finally {
				this.lockTreeMap.writeLock().unlock();
			}
		} catch (InterruptedException e) {
			log.error("take Messages exception", e);
		}

		return result;
	}

	public void clear() {
		try {
			this.lockTreeMap.writeLock().lockInterruptibly();
			try {
				this.msgTreeMap.clear();
				this.msgTreeMapTemp.clear();
				this.msgCount.set(0);
				this.queueOffsetMax = 0L;
			} finally {
				this.lockTreeMap.writeLock().unlock();
			}
		} catch (InterruptedException e) {
			log.error("rollback exception", e);
		}
	}

	public long getLastLockTimestamp() {
		return lastLockTimestamp;
	}

	public void setLastLockTimestamp(long lastLockTimestamp) {
		this.lastLockTimestamp = lastLockTimestamp;
	}

	public Lock getLockConsume() {
		return lockConsume;
	}

	public long getLastPullTimestamp() {
		return lastPullTimestamp;
	}

	public void setLastPullTimestamp(long lastPullTimestamp) {
		this.lastPullTimestamp = lastPullTimestamp;
	}

	public long getMsgAccCnt() {
		return msgAccCnt;
	}

	public void setMsgAccCnt(long msgAccCnt) {
		this.msgAccCnt = msgAccCnt;
	}

	public long getTryUnlockTimes() {
		return this.tryUnlockTimes.get();
	}

	public void incTryUnlockTimes() {
		this.tryUnlockTimes.incrementAndGet();
	}

	public void fillProcessQueueInfo(final ProcessQueueInfo info) {
		try {
			this.lockTreeMap.readLock().lockInterruptibly();

			if (!this.msgTreeMap.isEmpty()) {
				info.setCachedMsgMinOffset(this.msgTreeMap.firstKey());
				info.setCachedMsgMaxOffset(this.msgTreeMap.lastKey());
				info.setCachedMsgCount(this.msgTreeMap.size());
			}

			if (!this.msgTreeMapTemp.isEmpty()) {
				info.setTransactionMsgMinOffset(this.msgTreeMapTemp.firstKey());
				info.setTransactionMsgMaxOffset(this.msgTreeMapTemp.lastKey());
				info.setTransactionMsgCount(this.msgTreeMapTemp.size());
			}

			info.setLocked(this.locked);
			info.setTryUnlockTimes(this.tryUnlockTimes.get());
			info.setLastLockTimestamp(this.lastLockTimestamp);

			info.setDroped(this.dropped);
			info.setLastPullTimestamp(this.lastPullTimestamp);
			info.setLastConsumeTimestamp(this.lastConsumeTimestamp);
		} catch (Exception e) {
		} finally {
			this.lockTreeMap.readLock().unlock();
		}
	}

	public long getLastConsumeTimestamp() {
		return lastConsumeTimestamp;
	}

	public void setLastConsumeTimestamp(long lastConsumeTimestamp) {
		this.lastConsumeTimestamp = lastConsumeTimestamp;
	}
}
